package org.eclipse.moquette.spi.persistence;


import com.alibaba.fastjson.JSON;
import com.google.common.base.Strings;
import com.google.common.collect.Lists;
import org.eclipse.moquette.proto.messages.AbstractMessage;
import org.eclipse.moquette.spi.IMatchingCondition;
import org.eclipse.moquette.spi.IMessagesStore;
import org.eclipse.moquette.spi.ISessionsStore;
import org.eclipse.moquette.spi.impl.events.PublishEvent;
import org.eclipse.moquette.spi.impl.subscriptions.Subscription;
import org.eclipse.moquette.spi.impl.subscriptions.SubscriptionsStore;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import redis.clients.jedis.Jedis;

import java.nio.ByteBuffer;
import java.util.*;
import java.util.function.Function;
import java.util.stream.Collectors;


public class RedisStore extends SubscriptionsStore implements IMessagesStore, ISessionsStore {
    private static final Logger LOG = LoggerFactory.getLogger(RedisStore.class);

    private static final String CLIENT_MAILBOX_PREFIX = "client_mailbox_";
    private static final String SUBS_PREFIX = "client_subs_";
    private static final String TOPIC_SUBS_PREFIX = "topic_subs_";
    private static final String RETAINED_PREFIX = "retained_msg_";
    private static final String FLY_PREFIX = "fly_msg_";
    private static final String FLY_ID_PREFIX = "fly_msg_id_";
    private static final String UUID_PREFIX = "client_msg_uuid_";
    private static final String QOS2_PREFIX = "qos2_msg_";


    private Jedis jedis;


    private String getClientMailboxKey(String clientId) {
        return CLIENT_MAILBOX_PREFIX + clientId;
    }

    private String getSUBSKey(String clientId) {
        return SUBS_PREFIX + clientId;
    }

    private String getTopicSUBSKey(String clientId) {
        return TOPIC_SUBS_PREFIX + clientId;
    }

    private String getRetainedKey(String topic) {
        return RETAINED_PREFIX + topic;
    }

    private String getFlyKey(String topic) {
        return FLY_PREFIX + topic;
    }

    private String getFlyIdKey(String topic) {
        return FLY_ID_PREFIX + topic;
    }

    private String getQOS2MSGKey(String topic) {
        return QOS2_PREFIX + topic;
    }


    @Override
    public void initStore() {
        jedis = new Jedis("localhost", 6379);
    }

    @Override
    public void cleanRetained(String topic) {
        jedis.del(getRetainedKey(topic));
    }

    @Override
    public void storeRetained(String topic, ByteBuffer message, AbstractMessage.QOSType qos) {
        if (!message.hasRemaining()) {
            //clean the message from topic
            jedis.del(getRetainedKey(topic));
        } else {
            //store the message to the topic
            byte[] raw = new byte[message.remaining()];
            message.get(raw);
            StoredMessage msg = new StoredMessage(raw, qos, topic);
            jedis.hset(getRetainedKey(topic), msg.hashCode() + "", JSON.toJSONString(msg));
        }
    }

    @Override
    public Collection<StoredMessage> searchMatching(IMatchingCondition condition) {
        LOG.debug("searchMatching scanning all retained messages, presents are {}", 0);
        return Lists.newArrayList();
    }

    @Override
    public void storePublishForFuture(PublishEvent evt) {
        LOG.debug("storePublishForFuture store evt {}", evt);
        jedis.hset(getClientMailboxKey(evt.getClientID()), evt.getMessageID() + "", JSON.toJSONString(evt));
    }

    @Override
    public List<PublishEvent> listMessagesInSession(String clientID) {

        Map<String, String> msgMap = jedis.hgetAll(getClientMailboxKey(clientID));
        if (null == msgMap) {
            return Lists.newArrayList();
        }

        return msgMap.entrySet().stream().
                map(entry -> JSON.parseObject(entry.getValue(), PublishEvent.class))
                .sorted((e1, e2) -> e1.getMessageID() - e2.getMessageID())
                .collect(Collectors.toList());
    }

    @Override
    public void removeMessageInSession(String clientID, Integer messageID) {
        jedis.hdel(getClientMailboxKey(clientID), messageID + "");
    }

    @Override
    public void dropMessagesInSession(String clientID) {
        jedis.del(getClientMailboxKey(clientID));
    }

    @Override
    public void cleanInFlight(String clientID, int packetID) {
        String publishKey = String.format("%s%d", clientID, packetID);
        jedis.hdel(getFlyKey(clientID), publishKey);
        jedis.hdel(getFlyIdKey(clientID), packetID + "");
    }

    @Override
    public void addInFlight(PublishEvent evt, String clientID, int packetID) {
        String publishKey = String.format("%s%d", clientID, packetID);
        jedis.hset(clientID, publishKey, JSON.toJSONString(evt));
    }

    /**
     * Return the next valid packetIdentifer for the given client session.
     */
    @Override
    public int nextPacketID(String clientID) {
        Long id = jedis.incr(UUID_PREFIX + clientID);
        jedis.hset(getFlyIdKey(clientID), id + "", id + "");
        return id.intValue();
    }

    @Override
    public void removeSubscription(String topic, String clientID) {
        LOG.debug("removeSubscription topic filter: {} for clientID: {}", topic, clientID);
        jedis.srem(getSUBSKey(clientID), topic);
        jedis.hdel(getTopicSUBSKey(topic), clientID);
    }

    @Override
    public void addNewSubscription(Subscription newSubscription) {
        final String clientID = newSubscription.getClientId();
        final String topic = newSubscription.getTopicFilter();
        jedis.sadd(getSUBSKey(clientID), topic);
        jedis.hset(getTopicSUBSKey(topic), clientID, JSON.toJSONString(newSubscription));
    }

    public List<Subscription> getTopicSubscriptions(String topic) {
        return jedis.hgetAll(getTopicSUBSKey(topic)).entrySet().stream()
                .map(entry -> JSON.parseObject(entry.getValue(), Subscription.class))
                .collect(Collectors.toList());
    }

    @Override
    public void wipeSubscriptions(String clientID) {
        jedis.del(getSUBSKey(clientID));
        jedis.hgetAll(getSUBSKey(clientID)).keySet()
                .forEach(topic -> jedis.hdel(getTopicSUBSKey(topic), clientID));
    }

    @Override
    public void updateSubscriptions(String clientID, Set<Subscription> subscriptions) {
        if (null == subscriptions || Strings.isNullOrEmpty(clientID)) {
            return;
        }

    }

    public void updateSubscriptions(String clientID, Function<Subscription, Subscription> editor) {
        if (Strings.isNullOrEmpty(clientID)) {
            return;
        }

        Set<String> topics = jedis.smembers(getSUBSKey(clientID));
        if (null == topics) return;
        topics.forEach(topic -> {
            String jsonStr = jedis.hget(getTopicSUBSKey(topic), clientID);
            Subscription subscription = JSON.parseObject(jsonStr, Subscription.class);
            jsonStr = JSON.toJSONString(editor.apply(subscription));
            jedis.hset(getTopicSUBSKey(topic), clientID, jsonStr);
        });
    }

    @Override
    public boolean contains(String clientID) {
        return jedis.exists(getSUBSKey(clientID));
    }

    @Override
    public List<Subscription> listAllSubscriptions() {
        //TODO
        return null;
    }

    @Override
    public void close() {
        //To change body of implemented methods use File | Settings | File Templates.
    }

    @Override
    public void persistQoS2Message(String publishKey, PublishEvent evt) {
        LOG.debug("persistQoS2Message store pubKey {}, evt {}", publishKey, evt);
        jedis.set(getQOS2MSGKey(publishKey), JSON.toJSONString(evt));
    }

    @Override
    public void removeQoS2Message(String publishKey) {
        jedis.del(getQOS2MSGKey(publishKey));
    }

    @Override
    public PublishEvent retrieveQoS2Message(String publishKey) {
        return JSON.parseObject(jedis.get(getQOS2MSGKey(publishKey)), PublishEvent.class);
    }

}